from dagster import get_dagster_logger, job, op

logger = get_dagster_logger()


@op
def op_arg():
    return (2, 1)


@op
def op_add(arg):
    r""" 读入参数做加法 """
    x, y = arg
    ret = x + y
    logger.info(f"{x} + {y} = {ret}")
    return ret


@op
def op_sub(arg):
    x, y = arg
    ret = x - y
    logger.info(f"{x} - {y} = {ret}")
    return ret


@op
def op_mul(arg):
    x, y = arg
    ret = x * y
    logger.info(f"{x} * {y} = {ret}")
    return ret


@op
def op_div(arg):
    x, y = arg
    ret = x / y
    logger.info(f"{x} / {y} = {ret}")
    return ret


@job
def arithmetic():
    arg = op_arg()
    add = op_add(arg)
    sub = op_sub(arg)
    mul = op_mul(arg)
    div = op_div(arg)
    logger.info(f"arg={arg},add={add},sub={sub},mul={mul},div={div}")


if __name__ == "__main__":
    result = arithmetic.execute_in_process(
        run_config={"loggers": {"console": {"config": {"log_level": "info"}}}}
    )
